package com.loong.kafka.java.producer;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;

/**
 * Created by 梁浩峰 on 2016/9/10 14:31.
 */
public class LogProducer {
    private Producer<String,String> inner;

    public LogProducer() throws Exception {
        Properties properties=new Properties();
        properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));
        ProducerConfig config=new ProducerConfig(properties);
        inner = new Producer<String, String>(config);
    }

    public void send(String topicName,String message){
        if (topicName == null||message==null){
            return;
        }
        KeyedMessage<String, String> km=new KeyedMessage<String, String>(topicName, message);
        inner.send(km);
    }

    public void send(String topicName, Collection<String> message){
        if (topicName == null||message==null){
            return;
        }
        if(message.isEmpty()){
            return;
        }
        List<KeyedMessage<String,String>> kms=new ArrayList<KeyedMessage<String, String>>();
        for (String entry: message
             ) {
            KeyedMessage<String,String> km = new KeyedMessage<String, String>(topicName,entry);
            kms.add(km);
        }
        inner.send(kms);
    }


    public void close(){
        inner.close();
    }

}
